package com.tzy.rpc.transport.netty.client;/**
 * @description:
 * @Author HP
 * @create 2022/3/14 20:02
 */

import com.tzy.rpc.entity.RpcRequest;
import com.tzy.rpc.entity.RpcResponse;
import com.tzy.rpc.enumeration.RpcError;
import com.tzy.rpc.exception.RpcException;
import com.tzy.rpc.factory.SingletonFactory;
import com.tzy.rpc.loadbalancer.LoadBalancer;
import com.tzy.rpc.register.NacosServiceDiscovery;
import com.tzy.rpc.register.ServiceDiscovery;
import com.tzy.rpc.serializer.CommonSerializer;
import com.tzy.rpc.transport.RpcClient;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;

/**
 * @author HP
 * @create 2022-03-14 20:02
 * @desc Netty实现的客户端
 **/
public class NettyClient implements RpcClient {

    private static final Logger logger= LoggerFactory.getLogger(NettyClient.class);

    private CommonSerializer serializer;
    private final ServiceDiscovery serviceDiscovery;
    private final UnprocessedRequests unprocessedRequests;

    private static final EventLoopGroup group;
    private static final Bootstrap bootstrap;

    static {
        group = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class);
    }


    public NettyClient() {
        serviceDiscovery=new NacosServiceDiscovery(null);
        unprocessedRequests= SingletonFactory.getInstance(UnprocessedRequests.class);
    }

    public NettyClient(LoadBalancer loadBalancer) {
        this.serviceDiscovery = new NacosServiceDiscovery(loadBalancer);
        unprocessedRequests= SingletonFactory.getInstance(UnprocessedRequests.class);
    }

    @Override
    public CompletableFuture<RpcResponse> sendRequest(RpcRequest rpcRequest) {
        if (serializer == null) {
            logger.error("未设置序列化器");
            throw new RpcException(RpcError.SERIALIZER_NOT_FOUND);
        }
        CompletableFuture<RpcResponse> resultFuture = new CompletableFuture<>();
        try {
            //从Nacos获取提供对应服务的服务端地址
            InetSocketAddress inetSocketAddress = serviceDiscovery.lookupService(rpcRequest.getInterfaceName());
            Channel channel = ChannelProvider.get(inetSocketAddress, serializer);
            if(!channel.isActive()) {
                group.shutdownGracefully();
                logger.error("无法从Nacos中获取到的服务端地址中调用服务{}...",rpcRequest.getMethodName());
                return null;
            }
            //将新请求放入未处理完的请求中
            unprocessedRequests.put(rpcRequest.getRequestId(), resultFuture);
            //向服务端发请求，并设置监听，关于writeAndFlush()的具体实现可以参考：https://blog.csdn.net/qq_34436819/article/details/103937188
            channel.writeAndFlush(rpcRequest).addListener((ChannelFutureListener) future1 -> {
                if(future1.isSuccess()){
                    logger.info(String.format("客户端发送消息：%s", rpcRequest.toString()));
                }else {
                    logger.error("发送消息时有错误发生:", future1.cause());
                    future1.channel().close();
                    resultFuture.completeExceptionally(future1.cause());
                }
            });
        } catch (Exception e) {
            logger.error("发送消息时有错误发生: ", e.getMessage());
            unprocessedRequests.remove(rpcRequest.getRequestId());
            //interrupt()这里作用是给受阻塞的当前线程发出一个中断信号，让当前线程退出阻塞状态，好继续执行然后结束
            Thread.currentThread().interrupt();
        }
        return resultFuture;
    }

    @Override
    public void setSerializer(CommonSerializer serializer) {
        this.serializer=serializer;
    }
}
